package com.zh.amqp.rabbit;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import static com.zh.amqp.rabbit.mqmsg.MsgProducer.DEMO_EXCHANGE;
import static com.zh.amqp.rabbit.mqmsg.MsgProducer.DEMO_QUEUE;
import static com.zh.amqp.rabbit.mqmsg.MsgProducer.ROUTING_KEY;

/**
 * @author ZH
 * @date 20:27 2021/4/14
 * <p>
 * <p>
 * 如何保证顺序消费：
 * 破坏顺序的场景：
 * 1、一个queue、多个consumer去消费：consumer消费过程执行不确定
 * 2、一个queue、一个consumer多线程去消费
 * 方式一：（将场景1转化为场景2）：拆分多个queue，每个queue一个consumer，就是多一些queue而已
 * 方式二：（场景2）：限制为一个消费者，设置一个并发数，设置每次只推送一个消息
 * <code>
 * spring:
 * rabbitmq:
 * listener:
 * simple:
 * concurrency: 1 #一个并发数
 * prefetch: 1 #每次只推送一个消息
 * acknowledge-mode: manual
 * </code>
 * <p>
 * 如何防止重复消费：
 * 方式一：防重表（只消费一次）
 * 方式二：保证消费消息的业务是幂等的（消费多次但是幂等）
 * <p>
 * <p>
 * 消息积压如何处理：
 * 方式一：增加消费者的消费能力
 * 适当增加 prefetch 的数量，让消费端一次多接受一些消息；
 * 或者 增加多个消费者
 */
@Slf4j
@Configuration
public class RabbitConfig {
    @Bean
    public DirectExchange demoExchange() {
        return new DirectExchange(DEMO_EXCHANGE, true, false);
    }

    @Bean
    public Queue demoQueue() {
        return new Queue(DEMO_QUEUE, true);
    }

    @Bean
    public Binding demoQueueBinding() {
        return BindingBuilder.bind(demoQueue()).to(demoExchange()).with(ROUTING_KEY);
    }

    /**
     * 消息转换器
     */
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
//        return new SerializerMessageConverter();
//        return new SimpleMessageConverter();
    }

    /**
     * 生产端的发送消息回调
     * setConfirmCallback()和setReturnsCallback()的区别：
     *      setConfirmCallback 用于确认message是否被MQ broker 接收。
     *      setReturnsCallback 用于确认message是否被路由到Queue；这需要在声明交换机时开启mandatory标志。
     *
     * 死信队列
     * 死信队列的触发条件是满足其一(先要配置死信队列)：
     * 1.消息被消费者拒绝（basicReject/basicNack）且 requeue=false
     * 2.消息过期（设置了 x-message-ttl）
     * 3.队列达到最大长度（x-max-length）
     */
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(messageConverter());
//        rabbitTemplate.setReplyTimeout(); //设置接收响应消息的超时时间


        //exchange接受到消息 的回调
        //correlationData：消息的唯一标识；
        //ack：消息是否已经发送到交换机
        rabbitTemplate.setConfirmCallback((CorrelationData correlationData, boolean ack, String cause) -> {

            if (ack) {
                log.info("【MQ】消息发送成功，broker接收成功：correlationData = {}, ack = {}, cause = {}"
                        , correlationData, ack, cause);
            } else {
                log.info("【MQ】消息发送失败，broker接收失败：correlationData = {}, ack = {}, cause = {}"
                        , correlationData, ack, cause);
                //可以重试发送
            }

        });

        //exchange路由消息失败 的回调
        rabbitTemplate.setReturnsCallback((ReturnedMessage message) -> {
            log.warn("【MQ】消息已被Broker接收到，但是路由失败：message = {}, replyCode = {}, replyText = {}, routingKey = {}"
                    , message.getMessage(), message.getReplyCode(), message.getReplyText(), message.getRoutingKey());

            //一般的业务处理：日志记录一下错误，消息记录到错误表中，手动补偿就好了
        });

        //true表示接收路由不可达的消息，理解为setReturnsCallback的开关
        // 若无法路由到符合的队列时：
        // true：RabbitMQ会调用basic.return命令将消息返回给生产者
        // false：RabbitMQ会将消息丢弃。（不会进入死信队列）
        rabbitTemplate.setMandatory(true);

        return rabbitTemplate;
    }

}
